package com.nl.deal;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.nl.DealDataToSendMain;
import com.nl.redis.RedisCluster;
import com.nl.util.AES;
import com.nl.util.ConnectionManager;
import com.nl.util.StringUtil;

public class DealDataToSend implements Runnable {

	private static final Logger log = LoggerFactory.getLogger(DealDataToSend.class);

	private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	private SimpleDateFormat delaysdf = new SimpleDateFormat("yyyyMMddHHmmss");
	private String redisCfg;
	private int count = 0;
	private RedisCluster cluster;

	private List<SmsSender> senders = new ArrayList<>();
	private String sendSql;
	private String xzsendSql;
	private String senderNum;
	// private int expire = 604800;
	// private int xjexpire = 86400;
	private SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HHmmss");
	private String flag;

	public DealDataToSend(String redisCfg, String senderNum, String flag) {
		this.redisCfg = redisCfg;
		this.senderNum = senderNum;
		this.flag = flag;
	}

	@Override
	public void run() {
		log.info("deal data start!");
		try {
			cluster = new RedisCluster(redisCfg);
			sendSql = "insert into boss_sm_send(id,receiver,sender,content,result,"
					+ "create_date,modify_date,fetch_id,city_id,region_id"
					+ ",region_name,station_id,station_name,mgr_id,mgr_name,user_place,user_city,is_local,is_mark,"
					+ "county_id," + "city_ext_id,redis_daycount,mgr_type,send_type,mode_id)"
					+ "values(seq_bomc_cmpp.nextval,?,?,?,?,sysdate,sysdate,?,?,?,?,?,?,?,?,?,?,?,?," + "?,"
					+ "?,?,?,?,?)";
			xzsendSql = "insert into XZ_MSISDN_SEND(id,receiver,sender,content,result," + "create_date,modify_date)"
					+ "values(seq_bomc_cmpp.nextval,?,?,?,?,sysdate,sysdate)";

		} catch (IOException e1) {
			e1.printStackTrace();
		}

		while (true) {
			if (DataCache.dealData.isEmpty()) {
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					log.error("thread sleep error :" + e.getMessage());
					e.printStackTrace();
				}
				continue;
			}
			String msg = DataCache.dealData.poll();
			if (null != msg) {
				try{
					List<String> list = StringUtil.split(msg, ",");
					String enterTime = list.get(3);
					String receiveTime = list.get(4);//yyyyMMddHH24miss 86400000
					receiveTime = receiveTime.substring(0, 4) + "-" +receiveTime.substring(4, 6)+ "-" +receiveTime.substring(6, 8) + " "+ receiveTime.substring(8);
					enterTime = enterTime.substring(0, 4) + "-" +enterTime.substring(4, 6)+ "-" +enterTime.substring(6, 8) + " "+ enterTime.substring(8);
					Date enterDate = sdf1.parse(enterTime);
					Date receiveDate = sdf1.parse(receiveTime);
					if ((receiveDate.getTime()- enterDate.getTime())> Integer.parseInt(DealDataToSendMain.overTime)) {
						count++;
						if (count%10000 == 0 && count !=0) {
							log.info("overTime Data size:"+count);
						}
						continue;
					}
					deal( list,msg);
				} catch (Exception e) {
					log.error("deal data error,msg = {"+msg+"}",e);
				}
			}	
		}
	}

	private void deal(List<String> list,String msg) throws ParseException {
		
		String msisdn = list.get(0);
		String cityId = list.get(2);
		///黑名单和七日已发可以提前过滤，不依赖其他条件
		///过滤黑名单
		String stationId = list.get(1);
		String mgrIds = DataCache.stationMgrRelationMap.get(stationId);
		if (null != mgrIds && !"".equals(mgrIds)) {
				List<String> mgrIdCollection = StringUtil.split(mgrIds, ",");
				for (String mgrId: mgrIdCollection) {
					//徐州态势感知业务
					if ("1000001839".equals(mgrId)) {
//						if (!cluster.existKey(msisdn+"#"+mgrId)) {
							Connection conn=null;
//							String now = sdf.format(new Date());
							try {
								String decmsisdn =AES.decrypt(msisdn,DealDataToSendMain.KEY);
								decmsisdn = decmsisdn.replace(decmsisdn.substring(decmsisdn.length()-4), "0000");
								if (DataCache.xjMsisdns.containsKey(decmsisdn)) {
									log.info("xjMsisdn"+msg);
									conn = ConnectionManager.getInstance().getConnection();
									PreparedStatement pres = conn.prepareStatement(xzsendSql);
									pres.setObject(1,msisdn);
									pres.setObject(2,"");
									pres.setObject(3,"");
									pres.setObject(4,"-1");
									pres.execute();
									if (null != pres) {
										pres.close();
									}
									if (null != conn) {
										conn.close();
									}
								}
							} catch (Exception e) {
								log.error("-----------"+e.toString());
								e.printStackTrace();
							}
//						}
					}else {
						
					String groupIdDateString = DataCache.mgrGroupMap.get(mgrId);
					String groupId = "";
					String saveTime = "";
					Boolean sendGroupMsg = false;
					String groupSendKey = "";
					if(null != groupIdDateString && !"".equals(groupIdDateString)){
						String[] groupStringArray = groupIdDateString.split("#");
						if(groupStringArray.length == 2){
							groupId = groupStringArray[0];
							saveTime = groupStringArray[1];
							if(cluster.existKey(groupId+"#"+msisdn)){
								log.info("--pass----this mgr has group send msg, mgrId = "+mgrId+" , groupId = "+groupId);
								continue;
							}else{
								groupSendKey = groupId+"#"+msisdn;
								sendGroupMsg = true;
							}
						}
						
					}
						
							
						
					
					
					///过滤七日已发
					Map<String, String> map = cluster.hgetAll(msisdn+"#"+mgrId);
										log.debug("get key = "+msisdn+"#"+mgrId+",map = "+map);
										log.debug("mgrIdCollection = "+mgrIdCollection+",deal msg = "+msg);
					if (null == map || map.isEmpty()) {
												log.debug("is not sevenuser,mgrId = "+mgrId+",deal msg = "+msg);
						SmsTemplateAttr smsTemplateAttr = DataCache.smsTemplateAttrMap.get(mgrId);
						if (null == smsTemplateAttr) {
							//log.info("项目模板过滤-未通过,mgrId = "+mgrId);
							continue;
						}
						String now = sdf.format(new Date());
						String isLocal = list.get(7);
						String isSendLocal = smsTemplateAttr.getIs_local();
						String localIsCz = smsTemplateAttr.getLocal_is_cz();
						String nonLocalIsCz = smsTemplateAttr.getNonlocal_is_cz();
						///从原视图choose_file_data剥离的逻辑，短信审核状态、启停状态、发送时间、是否发送本地号码过滤
						///短信审核状态、启停状态、发送时间已在sql中做了筛选
						if (now.compareTo(smsTemplateAttr.getStart_date()) >= 0 
								&& now.compareTo(smsTemplateAttr.getEnd_date()) <= 0 
								&& now.compareTo(smsTemplateAttr.getStart_time()) >= 0 
								&& now.compareTo(smsTemplateAttr.getEnd_time()) <= 0 
								) {
							///&& isLocal.compareTo(isSendLocal) >= 0
							String isCz = list.get(8);
							
							if (isSendLocal.equals("2")) {
								isSendLocal = isLocal;
							}
							
							//IT传来的isLocal 0是本地，1是外地 和本地不一致则直接过滤掉
							if (!isLocal.equals(isSendLocal)) {
								log.debug("isLocal = "+isLocal+",isSendLocal = "+isSendLocal+", msgId = "+mgrId);
								continue;
							}
							
							//项目发本地，屏蔽常驻人口，号码为本地且为常驻人口，不发送短信
							if(isSendLocal.equals("0")&&isCz.equals("1")&&"0".equals(localIsCz)){
								log.debug("PASS LOCAL_IS_CZ----isLocal = "+isLocal+",isCz = "+isCz+", localIsCz = "+localIsCz);
								continue;
							}
							//项目发外地，发送常驻人口，号码为外地且为非常驻人口，不发送短信
							if(isSendLocal.equals("1")&&isCz.equals("0")&&"2".equals(nonLocalIsCz)){
								log.debug("PASS NONLOCAL_IS_CZ----isLocal = "+isLocal+",isCz = "+isCz+", nonLocalIsCz = "+nonLocalIsCz);
								continue;
							}
							//项目发外地，发送非常驻人口，号码为外地且为常驻人口，不发送短信
							if(isSendLocal.equals("1")&&isCz.equals("1")&&"0".equals(nonLocalIsCz)){
								log.debug("PASS NONLOCAL_IS_CZ----isLocal = "+isLocal+",isCz = "+isCz+", nonLocalIsCz = "+nonLocalIsCz);
								continue;
							}
							
							String isPronivce = smsTemplateAttr.getIs_pronivce();
							if (null !=isPronivce && !"".equals(isPronivce) && "0".equals(isPronivce)) {
								if (null==list.get(5) || "".equals(list.get(5))) {
									log.debug("pronivce:"+msisdn+","+stationId);
									continue;
								}
							}
							
							
							///剔除常驻人口
							//发本地时不管是否为常驻，不发本地时必须为非常驻人口。
							//if ("0".equals(isCz) || "0".equals(isSendLocal)) {
								
								ThresholdInfo info = DataCache.thresholdMap.get(mgrId);
								if (info == null) {
									continue;
								}

								String delayMin = smsTemplateAttr.getReal_time();

								///延迟发送
								if (null != delayMin && !"".equals(delayMin) && !"0".equals(delayMin)) {
									String enterTime = list.get(3);
									long nowTime = System.currentTimeMillis();
									long delayTime = delaysdf.parse(enterTime).getTime();
									long time = (nowTime - delayTime)/1000/60;
									long compareTime = Long.valueOf(delayMin);
									if(!msg.endsWith(",sourceSms")){
										msg = msg+",sourceSms";
									}
									if (time < compareTime && !DataCache.dealData.contains(msg)) {
										DataCache.dealData.add(msg);
										continue;
									}
								}

								StationMgrRelation stationMgrRelation = DataCache.stationMgrINfoRelationMap.get(mgrId+"#"+stationId);
								String regionId = null;
								String regionName = null;
								String stationName = null;
								String mgrName = null;
								String cityExtId = null;
								//String checkPath=null;

							if (null != stationMgrRelation) {
								regionId = stationMgrRelation.getRegion_id();
								regionName = stationMgrRelation.getRegion_name();
								stationName = stationMgrRelation.getStation_name();
								mgrName = stationMgrRelation.getName();
								cityExtId = stationMgrRelation.getCity_id();
								// checkPath=stationMgrRelation.getCheck_path();
							}
							String countyId = smsTemplateAttr.getCountyId();
							String mgrType = smsTemplateAttr.getMgrType();

							// 发送类型
							String sendType = smsTemplateAttr.getSendType();
							// 5g模板编码
							String modeId = smsTemplateAttr.getModeId();

							// 金坛茅山旅游2020V2项目过滤镇江号段
							String realMsisdn = null;
							String realMsisdnSub = null;

//							try {
//								realMsisdn = AES.decrypt(msisdn, DealDataToSendMain.KEY);
//								realMsisdnSub = realMsisdn.substring(0, 7);
//								if ("1000002426".equals(mgrId)) {
//									if (DataCache.blackUser.containsKey(realMsisdnSub + "#" + mgrId)) {
//										log.info("---------mgr:1000002426-------pass msisdn:" + realMsisdn);
//									}
//								}
//							} catch (Exception e) {
//								e.printStackTrace();
//							}
							
							try {
						          realMsisdn = AES.decrypt(msisdn, DealDataToSendMain.KEY);
						          realMsisdnSub = realMsisdn.substring(0, 7);
						        } catch (Exception e) {
						          e.printStackTrace();
						        }
							

							// 过滤黑名单
							if (DataCache.blackUser.containsKey(msisdn)
									|| DataCache.blackUser.containsKey(msisdn + "#" + cityId)
									|| DataCache.blackUser.containsKey(msisdn + "#" + countyId)
									|| DataCache.blackUser.containsKey(msisdn + "#" + mgrId)
									|| DataCache.blackUser.containsKey(realMsisdnSub)
									|| DataCache.blackUser.containsKey(realMsisdnSub + "#" + cityId)
									|| DataCache.blackUser.containsKey(realMsisdnSub + "#" + countyId)
									|| DataCache.blackUser.containsKey(realMsisdnSub + "#" + mgrId)) {
								continue;
							}

							String key = msisdn + "#" + mgrId;
							// String result = cluster.hmsetResult(key,
							// sevenMap);
							Long result = cluster.hsetNx(key, "INSERT_FLAG", flag);
							if ("0".equals(String.valueOf(result))) {
								// log.info("----key has already
								// exist!--msisdn+mgrId:"+key);
								continue;
							}
							cluster.hsetNx(key, "ARRIVAL_CITY", cityId);
							cluster.hsetNx(key, "SEND_TIME", now);
							cluster.hsetNx(key, "REGION_ID", regionId);
							cluster.hsetNx(key, "IS_LOCAL", isLocal);

								long timeCount=0;
								long dayCount=0;
								long weekCount=0;
								long monthCount=0;
								long quarterCount=0;
								long yearCount=0;
								long totalCount=0;
								//从redis中将项目数量加1然后取出项目的实时发送量
								try {
									timeCount=cluster.incrBy(mgrId+"timeCount", 1L);
									dayCount=cluster.incrBy(mgrId+"dayCount", 1L);
									weekCount=cluster.incrBy(mgrId+"weekCount", 1L);
									monthCount=cluster.incrBy(mgrId+"monthCount", 1L);
									quarterCount=cluster.incrBy(mgrId+"quarterCount", 1L);
									yearCount=cluster.incrBy(mgrId+"yearCount", 1L);
									totalCount=cluster.incrBy(mgrId+"totalCount", 1L);
									
									//根据项目的发送量过滤阀值
									if ((info.getTotalThreshold() != 0 && info.getTotalThreshold() < totalCount)
											|| (info.getCycleThreshold() != 0 && info.getCycle() == 1 && info.getCycleThreshold() < yearCount)
											|| (info.getCycleThreshold() != 0 && info.getCycle() == 2 && info.getCycleThreshold() < monthCount)
											|| (info.getCycleThreshold() != 0 && info.getCycle() == 3 && info.getCycleThreshold() < weekCount)
											|| (info.getCycleThreshold() != 0 && info.getCycle() == 4 && info.getCycleThreshold() < dayCount)
											|| (info.getCycleThreshold() != 0 && info.getCycle() == 6 && info.getCycleThreshold() < quarterCount)
											|| (info.getCycleThreshold() != 0 && info.getCycle() == 5 && info.getCycleThreshold() < timeCount))
									{
										try {
											cluster.decrBy(mgrId+"timeCount", 1L);
											cluster.decrBy(mgrId+"dayCount", 1L);
											cluster.decrBy(mgrId+"weekCount", 1L);
											cluster.decrBy(mgrId+"monthCount", 1L);
											cluster.decrBy(mgrId+"quarterCount", 1L);
											cluster.decrBy(mgrId+"yearCount", 1L);
											cluster.decrBy(mgrId+"totalCount", 1L);
											//log.info("项目发送量已经到达阀值,已做减一操作"+mgrId+"目前发送量为："+daycount);
										} catch (Exception e) {
											log.error("阀值已到，对redis中的发送量减一出现异常,项目Id:"+mgrId+e.toString());
											e.printStackTrace();
										}
										//发送量已到，将redis中key过期时间设置为0
										cluster.expire(key, 0);	
										continue;
									}
								} catch (Exception e) {
									log.error("增加redis中的发送量出现异常："+e.toString());
									e.printStackTrace();
								}
								
								cluster.hsetNx(key, "REDIS_DAYCOUNT", dayCount+"");
								
								
								SmsSender smsSender = new SmsSender();
								//redis数量标记
								smsSender.setRedisDayCount(dayCount+"");
								
								// 发送类型
								smsSender.setSendType(sendType);
								//项目模板  5g项目才有， 短信项目为空
								smsSender.setModeId(modeId);
								smsSender.setMgrType(mgrType);
								smsSender.setCountyId(countyId);
								smsSender.setReceiver(msisdn);
								smsSender.setSender(senderNum);
								//项目短信内容，5g项目短信内容为空
								String content = DataCache.msgContens.get(mgrId);
								smsSender.setContent(content==null?"5gContent":content);
								smsSender.setResult("-1");
								smsSender.setFetch_id("");
								smsSender.setCity_id(cityId);
								smsSender.setRegion_id(regionId == null ? "":regionId);
								smsSender.setRegion_name(regionName == null ? "":regionName);
								smsSender.setStation_id(stationId);
								smsSender.setStation_name(stationName == null ? "":stationName); 
								smsSender.setMgr_id(mgrId);
								smsSender.setMgr_name(mgrName == null ? "":mgrName);
								smsSender.setUser_place(list.get(5) == null ? "":list.get(5));
								smsSender.setUser_city(list.get(6) == null ? "":list.get(6));
								smsSender.setIs_local(isLocal);
								smsSender.setIs_mark("0");
								smsSender.setCity_ext_id(cityExtId == null ? "":cityExtId);
								//smsSender.setCheck_path(checkPath);
							
								//log.debug("set result = "+result+",key = "+key);
								if (!"0".equals(isLocal)) {
									//外地号码保存时间
									String nonlocalTime=smsTemplateAttr.getNonlocal_date();
									if (null != nonlocalTime && !"".equals(nonlocalTime)) {   
										int overExpire = Integer.valueOf(nonlocalTime)*24*60*60;
										
										cluster.expire(key, overExpire);
									}
								} else {
									String overSeven = smsTemplateAttr.getLocal_date();
									if (null != overSeven && !"".equals(overSeven)) {
										int overExpire = Integer.valueOf(overSeven)*24*60*60;
										cluster.expire(key, overExpire);
									}
								}
								
								//保存项目分组发送信息
								if(sendGroupMsg){
									cluster.set(groupSendKey, delaysdf.format(new Date())+"#"+saveTime);
									cluster.expire(groupSendKey,Integer.parseInt(saveTime)*24*60*60);
								}
								
								
								sendToDb(smsSender,cityId,now,regionId,isLocal,key,smsTemplateAttr);
								//					            log.debug("send matched data,deal msg = "+msg);
							} else {
								//log.debug("isCz = "+isCz+",isSendLocal = "+isSendLocal+", msgId = "+mgrId);
								//log.debug("isCz = "+isCz+",smsTemplateAttr.isCz = "+smsTemplateAttr.getIs_Cz()+", msgId = "+mgrId);
							}
						} 
					}
				}
			}
		}

	private void sendToDb(SmsSender sender,String cityId,String  now,String  regionId,String isLocal, String key,SmsTemplateAttr smsTemplateAttr) {
		senders.add(sender);
		Connection conn = null;
		try {
			if (senders.size() >= 50) {
//				log.debug("set send data to db!");
					conn = ConnectionManager.getInstance().getConnection();
					PreparedStatement pres = conn.prepareStatement(sendSql);
					for (SmsSender senderTmp: senders) {
						String msisdn = senderTmp.getReceiver();
						String tmpMgrId = senderTmp.getMgr_id();
						String redisDayCount=senderTmp.getRedisDayCount();

						String tmpFlag = cluster.hget(msisdn+"#"+tmpMgrId, "INSERT_FLAG");
						if (null != tmpFlag && !"".equals(tmpFlag) && !flag.equals(tmpFlag)) {
								
							log.info("Thread "+tmpFlag+" omit already send this msisdn, msisdn = "+msisdn+",mgrId"+tmpMgrId+",redisDayCount:"+redisDayCount);
							try {
								cluster.decrBy(tmpMgrId+"timeCount", 1L);
								cluster.decrBy(tmpMgrId+"dayCount", 1L);
								cluster.decrBy(tmpMgrId+"weekCount", 1L);
								cluster.decrBy(tmpMgrId+"monthCount", 1L);
								cluster.decrBy(tmpMgrId+"quarterCount", 1L);
								cluster.decrBy(tmpMgrId+"yearCount", 1L);
								cluster.decrBy(tmpMgrId+"totalCount", 1L);
								
							} catch (Exception e) {
								log.error("阀值已到，对redis中的发送量减一出现异常,项目Id:"+tmpMgrId+e.toString());
								e.printStackTrace();
							}
							
							continue;
						}

						if (!DataCache.mgrSend.containsKey(tmpMgrId)) {
							DataCache.mgrSend.put(tmpMgrId, 1);
						} else {
							int alreadySend = DataCache.mgrSend.get(tmpMgrId);
							DataCache.mgrSend.put(tmpMgrId, alreadySend+1);
						}
						pres.setObject(1, msisdn);
						pres.setObject(2, senderTmp.getSender());
						pres.setObject(3, senderTmp.getContent());
						pres.setObject(4, senderTmp.getResult());
						pres.setObject(5, DealDataToSendMain.master);
						pres.setObject(6, senderTmp.getCity_id());
						pres.setObject(7, senderTmp.getRegion_id());
						pres.setObject(8, senderTmp.getRegion_name());
						pres.setObject(9, senderTmp.getStation_id());
						pres.setObject(10, senderTmp.getStation_name());
						pres.setObject(11, tmpMgrId);
						pres.setObject(12, senderTmp.getMgr_name());
						pres.setObject(13, senderTmp.getUser_place());
						pres.setObject(14, senderTmp.getUser_city());
						pres.setObject(15, senderTmp.getIs_local());
						pres.setObject(16, senderTmp.getIs_mark());
						pres.setObject(17, senderTmp.getCountyId());
						pres.setObject(18, senderTmp.getCity_ext_id());
						pres.setObject(19, senderTmp.getRedisDayCount());
						pres.setObject(20, senderTmp.getMgrType());
						pres.setObject(21, senderTmp.getSendType());
						pres.setObject(22, senderTmp.getModeId());
						//pres.setObject(20, senderTmp.getCheck_path());
						pres.addBatch();
					}
					pres.executeBatch();
					if (null != pres) {
						pres.close();
					}
					if (null != conn) {
						conn.close();
					}
					senders.clear();
					//		         log.debug("set send data to db end!");
				
			}
		} catch (Exception e) {
			//发生异常，先清空list里数据防止重复入表发送，20211208
			senders.clear();
			log.error("----",e);
			if (null != conn) {
				try {
					conn.close();
				} catch (SQLException e1) {
					log.error("close conn failed");
					e1.printStackTrace();
				}
			}
		}
	}

}
